package com.guandata.cdc.model.db_type.mysql;

import com.guandata.cdc.model.source.AbstractSourceConfig;
import lombok.SneakyThrows;
import lombok.ToString;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.Optional;

/**
 * @author daishuyuan
 */
@ToString
public class MysqlSourceConfig extends AbstractSourceConfig {
    public final String sourceTable;

    public final String connector = "mysql-cdc";

    public final MysqlStartupMode startUpMode;

    public final String serverId;

    public final String sourceJdbcUrl;

    @SneakyThrows
    public MysqlSourceConfig(ParameterTool parameterTool) {
        super(parameterTool);
        this.sourceTable = String.format("%s.%s", sourceDatabase, parameterTool.getRequired("source.table"));
        this.startUpMode = Optional
                .ofNullable(parameterTool.get("source.startup.mode"))
                .map(MysqlStartupMode::fromCode)
                .orElse(MysqlStartupMode.INITIAL);
        this.serverId = parameterTool.getRequired("source.server.id");
        this.sourceJdbcUrl = String.format(MysqlConst.SHORT_JDBC_URL_PREFIX, sourceHostname, sourcePort);
    }
}
